package doit20.datayi.etl

import java.io.{ByteArrayOutputStream, DataOutputStream, FileOutputStream, ObjectOutputStream}
import java.nio.ByteBuffer

import org.apache.spark.sql.SparkSession
import org.roaringbitmap.RoaringBitmap

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-04-18
 * @desc app日志流量统计分析多维聚合基表
 */
object ApplogTafficDimensionAggrBase {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 读数据  app_tfc_aggr_base
    val tfcBase = spark.read.table("doit20dws.app_tfc_aggr_base").where("dt='2021-04-14'")

    /**
     * 自定义bitmap生成函数
     * 输入：guid列表
     * 返回：装着这些guid的bitmap对象的序列化结果
     */
    val bitmap: Array[Int] => Array[Byte] = (guids:Array[Int])=>{
      val bm = RoaringBitmap.bitmapOf(guids: _*)
      val baout = new ByteArrayOutputStream()
      val dataout = new DataOutputStream(baout)
      bm.serialize(dataout)

      baout.toByteArray
    }
    spark.udf.register("bitmap",bitmap)


    // 运算 + 输出
    tfcBase.createTempView("tfc")
    spark.sql(
      """
        |
        |insert into table doit20dws.app_tfc_dimension_aggr_base partition(dt='2021-04-14')
        |select
        |   appid         ,
        |   appversion    ,
        |   carrier       ,
        |   devicetype    ,
        |   ip            ,
        |   nettype       ,
        |   osname        ,
        |   osversion     ,
        |   releasechannel,
        |   resolution    ,
        |   hour          ,
        |   province      ,
        |   city          ,
        |   region        ,
        |   isnew         ,
        |   start_page    ,
        |   end_page      ,
        |   bitmap(collect_set(guid)) as uv_bitmap,
        |   sum(pv_cnt)                as pv_cnt,
        |   sum(end_ts - start_ts )    as acctime_amt
        |from
        |  tfc
        |group by
        |   appid         ,
        |   appversion    ,
        |   carrier       ,
        |   devicetype    ,
        |   ip            ,
        |   nettype       ,
        |   osname        ,
        |   osversion     ,
        |   releasechannel,
        |   resolution    ,
        |   hour          ,
        |   province      ,
        |   city          ,
        |   region        ,
        |   isnew         ,
        |   start_page    ,
        |   end_page
        |""".stripMargin)

    spark.close()

  }

}
